package com.sdk.bigdata.client;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.cronutils.model.CronType;
import com.sdk.bigdata.constant.Constants;
import com.sdk.bigdata.executor.CronScheduledThreadPoolExecutor;
import com.sdk.bigdata.http.BigDataHttpRequest;
import com.sdk.bigdata.http.BigDataHttpResponseResult;
import com.sdk.bigdata.model.*;
import com.sdk.bigdata.utils.LoggerUtil;
import com.sdk.bigdata.utils.OkHttpUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

class DataConsumer {

    private CronScheduledThreadPoolExecutor cronScheduledCronThreadPoolExecutor;
    private ExecutorService fixedExecutor;
    private DataConfig config;

    // 刷新标记，用来标识定时任务将缓冲的数据进行flush消费
    private volatile boolean freshFlag = false;
    // 批量缓冲数据线程是否循环标记
    private volatile boolean loopShutdownFlag = false;

    // 批量的数据大小
    private AtomicInteger batchCount;
    // 缓冲队列
    private ConcurrentLinkedQueue<List<BaseModel>> bufferQueue;

    // 用queueSize来充当 bufferQueue的容器大小来充当blocking的作用
    private AtomicInteger queueSize;

    // 用来记录已经处理的数据量,如果数值超过100亿亿，会重置为0
    private AtomicLong total;

    private Logger logger;
    DataConsumer(DataConfig config) {
        this.config = config;
        this.batchCount = new AtomicInteger(0);
        // 初始化maxConsumer个线程 + 1个批量缓冲数据的线程
        this.fixedExecutor = Executors.newFixedThreadPool(config.getMaxConsumer() + 1);

        // 初始化定时线程(防止数据缓冲堆积，数据不上报)
        this.cronScheduledCronThreadPoolExecutor = new CronScheduledThreadPoolExecutor(1);
        cronScheduledCronThreadPoolExecutor.scheduleWithCron(new CronConsumerRunnable(), config.getCronSpec(), CronType.QUARTZ);
        bufferQueue = new ConcurrentLinkedQueue<>();

        // 拉取数据
        fixedExecutor.execute(new BatchRunnable());

        //与blockingQueue的size相对应
        queueSize = new AtomicInteger(0);

        total = new AtomicLong(0);

        logger = LoggerFactory.getLogger(DataConsumer.class);
    }

    private class BatchRunnable implements Runnable {
        @Override
        public void run() {
            List<BaseModel> baseModelList = new ArrayList<>(config.getMaxSize());
            while (true) {
                final BaseModel baseModel;
                if (!loopShutdownFlag) {
                    try {
                        boolean offerFlag = false;
                        if (freshFlag) {
                            if (!baseModelList.isEmpty()) {
                                offerFlag = true;
                            }
                            freshFlag = false;
                        }
                        if (!offerFlag
                                && ((baseModel = DataQueue.getQueue().poll(50, TimeUnit.MILLISECONDS)) != null)) {
                            baseModelList.add(baseModel);
                            if (batchCount.incrementAndGet() >= config.getMaxSize()) {
                                offerFlag = true;
                            }
                        }
                        if (offerFlag) {
                            bufferQueue.offer(baseModelList);
                            queueSize.incrementAndGet();
                            fixedExecutor.execute(new ConsumerRunnable());
                            batchCount = new AtomicInteger(0);
                            baseModelList = new ArrayList<>(config.getMaxSize());
                        }
                    } catch (InterruptedException e) {
                        LoggerUtil.log(logger, Constants.ERROR, e.getMessage());
                    }
                } else {
                    if (!baseModelList.isEmpty()) {
                        flushForData(baseModelList);
                        List<BaseModel> remainingList = getCurrentQueueAndEnsure(DataQueue.getQueue().size());
                        if (!remainingList.isEmpty()) {
                            flushForData(baseModelList);
                        }
                    }
                    break;
                }
            }
        }
    }

    /**
     * 消费线程，从阻塞队列拿数据进行上报
     */
    private class ConsumerRunnable implements Runnable {
        @Override
        public void run() {
            List<BaseModel> currentBaseModelList;
            while ((currentBaseModelList = bufferQueue.poll()) != null) {
                LoggerUtil.log(logger, Constants.DEBUG, "批次获取的数据量为:" + currentBaseModelList.size() + ";数据为:" + JSONObject.toJSONString(currentBaseModelList));
                queueSize.decrementAndGet();
                flushForData(currentBaseModelList);
            }
        }
    }

    /**
     * 定时线程，定时上报数据
     */
    private class CronConsumerRunnable implements Runnable {
        @Override
        public void run() {
            List<BaseModel> currentBaseModelList = getCurrentQueueAndEnsure(config.getMaxSize());
            try {
                LoggerUtil.log(logger, Constants.INFO, "此时已经处理的数据量为:" + total.get());
                LoggerUtil.log(logger, Constants.DEBUG, "定时任务执行获取的数据量为：" + currentBaseModelList.size() + ";数据为:" + JSONObject.toJSONString(currentBaseModelList));
            } catch (Exception ignore) {
            }
            if (!currentBaseModelList.isEmpty()) {
                flushForData(currentBaseModelList);
            } else {
                freshFlag = true;
            }
        }

    }

    /**
     * 处理数据
     *
     * @param messageList 数据
     */
    private void flushForData(List<BaseModel> messageList) {
        LoggerUtil.log(logger, Constants.DEBUG, "flushForData,数据处理中,数据量为:" + messageList.size() + ";数据为" + JSONObject.toJSONString(messageList));
        BigDataHttpResponseResult result = OkHttpUtils.doPost(buildHttpRequest(messageList), config.getUrl(), config.getTimeout().intValue() * 1000);
        messageList.clear();
        try {
            if (null != result && null != result.getData() && null != result.getData().get("success")) {
                int success = Integer.parseInt(result.getData().get("success"));
                // 如果数值达到100亿亿，则进行清空
                if (total.get() >= 1000000000000000000L) {
                    LoggerUtil.log(logger, Constants.INFO, "total初始化中，此刻的total值为:" + total.get() + ",将要初始化为0");
                    total.set(0);
                }
                total.addAndGet(success);
            }
        } catch (Exception ignore) {
        }
        LoggerUtil.log(logger, Constants.DEBUG, "flushForData处理完成,result=" + result);
    }

    /**
     * 组装httpRequest
     *
     * @param messageList 数据
     * @return BigDataHttpRequest
     */
    private BigDataHttpRequest buildHttpRequest(List<BaseModel> messageList) {
        BigDataHttpRequest request = new BigDataHttpRequest();
        request.setApikey(config.getSecret());
        request.setGameId(config.getGameId());
        JSONArray jsonArray = new JSONArray();
        for (BaseModel baseModel : messageList) {
            jsonArray.add(JSONObject.parse(JSON.toJSONString(baseModel)));
        }
        request.setDataList(jsonArray);
        request.setCompression(config.getCompression());
        request.setTimezone(config.getTimezone());
        return request;
    }

    /**
     * 获取当前queue的快照
     *
     * @return 当前queue
     */
    private List<BaseModel> getCurrentQueueAndEnsure(Integer maxSize) {
        final List<BaseModel> currentBaseModelList = new ArrayList<>();
        BaseModel baseModel;
        for (int i = 0; i < maxSize; i++) {
            if ((baseModel = DataQueue.getQueue().poll()) != null) {
                currentBaseModelList.add(baseModel);
            } else {
                break;
            }
        }
        return currentBaseModelList;
    }

    /**
     * 获取当前队列大小
     *
     * @return 队列大小
     */
    Integer getQueueSize() {
        return queueSize.get();
    }

    /**
     * 关闭线程且清空缓冲
     */
    void close() {
        LoggerUtil.log(logger, Constants.INFO, "开始关闭服务");
        // 拒绝接入数据
        DataQueue.shutdownFlag = true;
        // 终止所有任务
        LoggerUtil.log(logger,Constants.INFO, "开始关闭定时线程池");

        cronScheduledCronThreadPoolExecutor.shutdown();
        try {
            cronScheduledCronThreadPoolExecutor.awaitTermination(config.getTimeout() * 3 + 2, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            LoggerUtil.log(logger,Constants.ERROR, "关闭定时线程池中断:" + e);
        }
        LoggerUtil.log(logger,Constants.INFO, "开始关闭消费线程池");
        loopShutdownFlag = true;
        fixedExecutor.shutdown();
        try {
            fixedExecutor.awaitTermination(getQueueSize() * (config.getTimeout() + 2), TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            LoggerUtil.log(logger,Constants.ERROR, "关闭消费线程池中断: " + e);
        }
        LoggerUtil.log(logger,Constants.INFO, "关闭消费线程池完成");
    }
}
